package com.mango.ch12;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.clustering.KMeans;
import org.apache.spark.mllib.clustering.KMeansModel;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.linalg.Vector;

import java.util.regex.Pattern;

public class SparkKmeansJob {
    static String inputPath = null;

    static {
        inputPath = "/Kmeans.csv";
//        inputPath = SparkKmeansJob.class.getResource("Kmeans.csv").getPath();
    }

    /**
     * 自定义map 的function实现类
     */
    private static class ParsePoint implements Function<String, Vector> {
        private static final Pattern SPACE = Pattern.compile(",");// 创建匹配模式

        @Override
        public Vector call(String s) throws Exception {
            String[] tok = SPACE.split(s);
            double[] point = new double[3];
            for (int i = 0; i < 3; i++) {
                point[i] = Double.valueOf(tok[i]);
            }
            //将point数组转换为vetor集合
            return Vectors.dense(point);
        }
    }

    public static void main(String[] args) {
        int k = 4;
        int iteration = 10;
        int runs = 1;
        SparkConf conf = new SparkConf().setAppName("JavaKmeanss");
        conf.setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> lines = jsc.textFile(inputPath, 2).coalesce(3);
        JavaRDD<Vector> points = lines.map(new Function<String, Vector>() {
            @Override
            public Vector call(String v1) throws Exception {
                String[] tok = v1.split(",");
                double[] point = new double[3];
                for (int i = 0; i < 3; i++) {
                    point[i] = Double.valueOf(tok[i]);
                }
                //将point数组转换为vetor集合
                return Vectors.dense(point);
            }
        });
        points.saveAsTextFile("/output");
        KMeansModel model = KMeans.train(points.rdd(), k, iteration, runs, KMeans.K_MEANS_PARALLEL());
        System.out.println("Cluster centers:");
        for (Vector center :
                model.clusterCenters()) {
            System.out.println(" " + center);

        }
        double cost = model.computeCost(points.rdd());
        System.out.println("Cost:" + cost);
        jsc.close();
        jsc.stop();
    }
}
